/******************************************************************************/
/* @ All Rights Reserved Copyright (C) 2014 broada,Ltd.                       */
/*                                                                            */
/*  NAME      = kafka_request.c                                               */
/*  NOTE      = kafka请求解析                                                 */
/*  DATE      = 2015/03/19 by zhengxd                                         */
/******************************************************************************/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdint.h>

#include "kafka_core.h"
#include "kafka_comm.h"
#include "kafka_request.h"

/*******************************************************************************
*  FUNC     :  解析produce request
*  ARGS     :  p_st_conn (入力链接信息)
*           :  p_st_hash_report(Frreport_data)
*           :  p_st_private(KAFKA私有数据)
*  RTN      :
*  NOTE     :
*******************************************************************************/
static int __kafka_report_req(Conntrack *p_st_conn
                               , Frhash_header *p_st_hash_report
                               , L7session_kafka_private *p_st_private)
{
    int i_ret = FR_SUCC;
    Frreport_data *p_st_report_data = NULL;
    Frreport_body_data *p_st_sess_body = NULL;
    L7_req_kafka *p_st_req = p_st_private->p_st_req;
    L7_fragment_kafka *p_st_fragment = p_st_private->p_st_fragment;
    p_st_req->ul_req_sequence = frct_seq_increase(p_st_conn);
    Frlist_node *p_st_node_topic = NULL;
    Frlist_node *p_st_node_partition = NULL;
    Frlist_node *p_st_node_message_set = NULL;
    Kafka_topic_info *p_st_topic_req = NULL;
    Kafka_partition_info *p_st_partition = NULL;
    Kafka_message_set_info *p_st_message_set = NULL;
    char arr_title[128] = {0};

    switch (p_st_req->i_req_type)
    {
        case KAFKA_REQ_TP_PRODUCE:
        case KAFKA_REQ_TP_MULTIPRODUCE:
            p_st_node_topic = FRLIST_FIRST_NODE(&p_st_req->st_topics);
            while (p_st_node_topic)
            {
                p_st_topic_req = (Kafka_topic_info *)FRLIST_GET_NODE_DATA(p_st_node_topic);

                p_st_node_partition = FRLIST_FIRST_NODE(&p_st_topic_req->st_partitions);
                while (p_st_node_partition)
                {
                    p_st_partition = (Kafka_partition_info *)FRLIST_GET_NODE_DATA(p_st_node_partition);

                    p_st_node_message_set = FRLIST_FIRST_NODE(&p_st_partition->st_messages);
                    while (p_st_node_message_set)
                    {
                        p_st_message_set = (Kafka_message_set_info *)FRLIST_GET_NODE_DATA(p_st_node_message_set);
                        /* 创建report body */
                        p_st_sess_body = frreport_body_new_l7(p_st_conn);
                        if (p_st_sess_body == NULL)
                        {
                            return APM_L7_ANALYZE_FAIL;
                        }
                        /* 追加connection相关信息 */
                        i_ret = frreport_body_l7_add_connInfo(p_st_sess_body, p_st_conn);
                        if (i_ret != FR_SUCC)
                        {
                            frreport_body_free(&p_st_sess_body);
                            return APM_L7_ANALYZE_FAIL;
                        }
                        /* add session topic */
                        i_ret = frreport_body_l7_add_str(p_st_sess_body
                                                         , KAFKA_REPORT_KEY_SESSION_TOPIC
                                                         , p_st_topic_req->p_topic_name ? p_st_topic_req->p_topic_name : "");
                        if (i_ret != FR_SUCC)
                        {
                            frreport_body_free(&p_st_sess_body);
                            return APM_L7_ANALYZE_FAIL;
                        }

                        /* add sequence id */
                        i_ret = frreport_body_l7_add_ulong(p_st_sess_body
                                                          , APM_REPORT_KEY_REQ_ID
                                                          , p_st_req->ul_req_sequence);
                        if (i_ret != FR_SUCC)
                        {
                            frreport_body_free(&p_st_sess_body);
                            return APM_L7_ANALYZE_FAIL;
                        }

                        /* add isResponseRequired */
                        if (p_st_req->i_need_response == FR_YES)
                        {
                            i_ret = frreport_body_l7_add_int(p_st_sess_body
                                                            , APM_REPORT_KEY_REQ_ISRESPREQUIRED, 1);

                        }
                        else
                        {
                            i_ret = frreport_body_l7_add_int(p_st_sess_body
                                                            , APM_REPORT_KEY_REQ_ISRESPREQUIRED, 0);
                        }
                        if (i_ret != FR_SUCC)
                        {
                            frreport_body_free(&p_st_sess_body);
                            return APM_L7_ANALYZE_FAIL;
                        }
                        /* add request start time */
                        p_st_req->st_start_timestamp = p_st_fragment->st_start_timestamp;
                        i_ret = frreport_body_l7_add_timeval(p_st_sess_body
                                                            , APM_REPORT_KEY_REQ_START_TIMESTAMP
                                                            , &p_st_req->st_start_timestamp);
                        if (i_ret != FR_SUCC)
                        {
                            frreport_body_free(&p_st_sess_body);
                            return APM_L7_ANALYZE_FAIL;
                        }
                        /* add request end time */
                        p_st_req->st_end_timestamp = p_st_fragment->st_end_timestamp;
                        i_ret = frreport_body_l7_add_timeval(p_st_sess_body
                                                            , APM_REPORT_KEY_REQ_END_TIMESTAMP
                                                            , &p_st_req->st_end_timestamp);
                        if (i_ret != FR_SUCC)
                        {
                            frreport_body_free(&p_st_sess_body);
                            return APM_L7_ANALYZE_FAIL;
                        }
                        /* add send time */
                        p_st_req->ul_send_time_us = kafka_get_interval_time(&p_st_fragment->st_start_timestamp
                                                                          , &p_st_fragment->st_end_timestamp);
                        i_ret = frreport_body_l7_add_ulong(p_st_sess_body
                                                          , APM_REPORT_KEY_REQ_SEND_TIME_US
                                                          , p_st_req->ul_send_time_us);
                        if (i_ret != FR_SUCC)
                        {
                            frreport_body_free(&p_st_sess_body);
                            return APM_L7_ANALYZE_FAIL;
                        }
                        /* add send data bytes */
                        p_st_req->ul_request_bytes = p_st_fragment->ui_total_len;
                        i_ret = frreport_body_l7_add_ulong(p_st_sess_body
                                                          , APM_REPORT_KEY_REQ_DATA_BYTES
                                                          , p_st_req->ul_request_bytes);
                        if (i_ret != FR_SUCC)
                        {
                            frreport_body_free(&p_st_sess_body);
                            return APM_L7_ANALYZE_FAIL;
                        }

                        if (p_st_topic_req->p_topic_name)
                        {
                            snprintf(arr_title, sizeof(arr_title), "PUT %s", p_st_topic_req->p_topic_name);
                        }
                        else
                        {
                            snprintf(arr_title, sizeof(arr_title), "PUT");
                        }
                        i_ret = frreport_body_l7_add_str(p_st_sess_body
                                                         , APM_REPORT_KEY_TRANS_TITLE
                                                         , arr_title);
                        if (i_ret != FR_SUCC)
                        {
                            frreport_body_free(&p_st_sess_body);
                            return APM_L7_ANALYZE_FAIL;
                        }

                        /* add request topic */
                        i_ret = frreport_body_l7_add_str(p_st_sess_body
                                                        , KAFKA_REPORT_KEY_REQ_TOPIC
                                                        , p_st_topic_req->p_topic_name ? p_st_topic_req->p_topic_name : "");
                        if (i_ret != FR_SUCC)
                        {
                            frreport_body_free(&p_st_sess_body);
                            return APM_L7_ANALYZE_FAIL;
                        }

                        /* add request method */
                        i_ret = frreport_body_l7_add_str(p_st_sess_body
                                                        , KAFKA_REPORT_KEY_REQ_METHOD
                                                        , "PUT");
                        if (i_ret != FR_SUCC)
                        {
                            frreport_body_free(&p_st_sess_body);
                            return APM_L7_ANALYZE_FAIL;
                        }

                        /* add request type */
                        i_ret = frreport_body_l7_add_str(p_st_sess_body
                                                        , KAFKA_REPORT_KEY_REQ_TYPE
                                                        , "text");
                        if (i_ret != FR_SUCC)
                        {
                            frreport_body_free(&p_st_sess_body);
                            return APM_L7_ANALYZE_FAIL;
                        }

                        /* add request message */
                        i_ret = frreport_body_l7_add_str(p_st_sess_body
                                                        , KAFKA_REPORT_KEY_REQ_TEXT
                                                        , p_st_message_set->p_value ? p_st_message_set->p_value : "");
                        if (i_ret != FR_SUCC)
                        {
                            frreport_body_free(&p_st_sess_body);
                            return APM_L7_ANALYZE_FAIL;
                        }

                        /* 创建report data header */
                        p_st_report_data = frreport_data_l7_new(FRREPORT_TYPE_L7DATA_REQ);
                        if (p_st_report_data == NULL)
                        {
                            frreport_body_free(&p_st_sess_body);
                            return APM_L7_ANALYZE_FAIL;
                        }
                        /* add body into header */
                        i_ret = frreport_data_add_body(p_st_report_data, p_st_sess_body);
                        if (i_ret != FR_SUCC)
                        {
                            frreport_body_free(&p_st_sess_body);
                            frreport_data_free(&p_st_report_data);
                            return APM_L7_ANALYZE_FAIL;
                        }

                        /* 上报request数据 */
                        i_ret = frreport_data_hash(p_st_hash_report, p_st_report_data);
                        if (i_ret != FR_SUCC)
                        {
                            frreport_data_free(&p_st_report_data);
                            return APM_L7_ANALYZE_FAIL;
                        }
                        p_st_node_message_set = FRLIST_NEXT_NODE(p_st_node_message_set);
                    }
                    p_st_node_partition = FRLIST_NEXT_NODE(p_st_node_partition);
                }
                p_st_node_topic = FRLIST_NEXT_NODE(p_st_node_topic);
            }
            break;

        case KAFKA_REQ_TP_FETCH:
        case KAFKA_REQ_TP_MULTIFETCH:
            p_st_node_topic = FRLIST_FIRST_NODE(&p_st_req->st_topics);
            while (p_st_node_topic)
            {
                p_st_topic_req = (Kafka_topic_info *)FRLIST_GET_NODE_DATA(p_st_node_topic);

                /* 创建report body */
                p_st_sess_body = frreport_body_new_l7(p_st_conn);
                if (p_st_sess_body == NULL)
                {
                    return APM_L7_ANALYZE_FAIL;
                }
                /* 追加connection相关信息 */
                i_ret = frreport_body_l7_add_connInfo(p_st_sess_body, p_st_conn);
                if (i_ret != FR_SUCC)
                {
                    frreport_body_free(&p_st_sess_body);
                    return APM_L7_ANALYZE_FAIL;
                }
                /* add session topic */
                i_ret = frreport_body_l7_add_str(p_st_sess_body
                                                , KAFKA_REPORT_KEY_SESSION_TOPIC
                                                , p_st_topic_req->p_topic_name ? p_st_topic_req->p_topic_name : "");
                if (i_ret != FR_SUCC)
                {
                    frreport_body_free(&p_st_sess_body);
                    return APM_L7_ANALYZE_FAIL;
                }

                /* add sequence id */
                i_ret = frreport_body_l7_add_ulong(p_st_sess_body
                                                  , APM_REPORT_KEY_REQ_ID
                                                  , p_st_req->ul_req_sequence);
                if (i_ret != FR_SUCC)
                {
                    frreport_body_free(&p_st_sess_body);
                    return APM_L7_ANALYZE_FAIL;
                }
                /* add isResponseRequired */
                i_ret = frreport_body_l7_add_int(p_st_sess_body
                                                , APM_REPORT_KEY_REQ_ISRESPREQUIRED, 1);
                if (i_ret != FR_SUCC)
                {
                    frreport_body_free(&p_st_sess_body);
                    return APM_L7_ANALYZE_FAIL;
                }
                /* add request start time */
                p_st_req->st_start_timestamp = p_st_fragment->st_start_timestamp;
                i_ret = frreport_body_l7_add_timeval(p_st_sess_body
                                                    , APM_REPORT_KEY_REQ_START_TIMESTAMP
                                                    , &p_st_req->st_start_timestamp);
                if (i_ret != FR_SUCC)
                {
                    frreport_body_free(&p_st_sess_body);
                    return APM_L7_ANALYZE_FAIL;
                }
                /* add request end time */
                p_st_req->st_end_timestamp = p_st_fragment->st_end_timestamp;
                i_ret = frreport_body_l7_add_timeval(p_st_sess_body
                                                    , APM_REPORT_KEY_REQ_END_TIMESTAMP
                                                    , &p_st_req->st_end_timestamp);
                if (i_ret != FR_SUCC)
                {
                    frreport_body_free(&p_st_sess_body);
                    return APM_L7_ANALYZE_FAIL;
                }
                /* add send time */
                p_st_req->ul_send_time_us = kafka_get_interval_time(&p_st_fragment->st_start_timestamp
                                                                  , &p_st_fragment->st_end_timestamp);
                i_ret = frreport_body_l7_add_ulong(p_st_sess_body
                                                  , APM_REPORT_KEY_REQ_SEND_TIME_US
                                                  , p_st_req->ul_send_time_us);
                if (i_ret != FR_SUCC)
                {
                    frreport_body_free(&p_st_sess_body);
                    return APM_L7_ANALYZE_FAIL;
                }
                /* add send data bytes */
                p_st_req->ul_request_bytes = p_st_fragment->ui_total_len;
                i_ret = frreport_body_l7_add_ulong(p_st_sess_body
                                                  , APM_REPORT_KEY_REQ_DATA_BYTES
                                                  , p_st_req->ul_request_bytes);
                if (i_ret != FR_SUCC)
                {
                    frreport_body_free(&p_st_sess_body);
                    return APM_L7_ANALYZE_FAIL;
                }

                if (p_st_topic_req->p_topic_name)
                {
                    snprintf(arr_title, sizeof(arr_title), "GET %s", p_st_topic_req->p_topic_name);
                }
                else
                {
                    snprintf(arr_title, sizeof(arr_title), "GET");
                }
                i_ret = frreport_body_l7_add_str(p_st_sess_body
                                                 , APM_REPORT_KEY_TRANS_TITLE
                                                 , arr_title);
                if (i_ret != FR_SUCC)
                {
                    frreport_body_free(&p_st_sess_body);
                    return APM_L7_ANALYZE_FAIL;
                }

                /* add request topic */
                i_ret = frreport_body_l7_add_str(p_st_sess_body
                                                , KAFKA_REPORT_KEY_REQ_TOPIC
                                                , p_st_topic_req->p_topic_name ? p_st_topic_req->p_topic_name : "");
                if (i_ret != FR_SUCC)
                {
                    frreport_body_free(&p_st_sess_body);
                    return APM_L7_ANALYZE_FAIL;
                }

                /* add request method */
                i_ret = frreport_body_l7_add_str(p_st_sess_body
                                                , KAFKA_REPORT_KEY_REQ_METHOD
                                                , "GET");
                if (i_ret != FR_SUCC)
                {
                    frreport_body_free(&p_st_sess_body);
                    return APM_L7_ANALYZE_FAIL;
                }

                /* 创建report data header */
                p_st_report_data = frreport_data_l7_new(FRREPORT_TYPE_L7DATA_REQ);
                if (p_st_report_data == NULL)
                {
                    frreport_body_free(&p_st_sess_body);
                    return APM_L7_ANALYZE_FAIL;
                }
                /* add body into header */
                i_ret = frreport_data_add_body(p_st_report_data, p_st_sess_body);
                if (i_ret != FR_SUCC)
                {
                    frreport_body_free(&p_st_sess_body);
                    frreport_data_free(&p_st_report_data);
                    return APM_L7_ANALYZE_FAIL;
                }

                /* 上报request数据 */
                i_ret = frreport_data_hash(p_st_hash_report, p_st_report_data);
                if (i_ret != FR_SUCC)
                {
                    frreport_data_free(&p_st_report_data);
                    return APM_L7_ANALYZE_FAIL;
                }
                p_st_node_topic = FRLIST_NEXT_NODE(p_st_node_topic);
            }
            break;

        default:
            return APM_L7_ANALYZE_SUCC;
    }

    return APM_L7_ANALYZE_SET_REQ(APM_L7_ANALYZE_SUCC);
}

/*******************************************************************************
*  FUNC     :  解析produce request
*  ARGS     :  p_st_private (KAFKA私有数据)
*  RTN      :  APM_L7_ANALYZE_SUCC
*              APM_L7_ANALYZE_FAIL
*  NOTE     :
*******************************************************************************/
int __kafka_produce_req(L7session_kafka_private *p_st_private)
{
    int i_ret = APM_L7_ANALYZE_SUCC;
    uint16_t u16_required_acks = 0;
    int i_topic_count = 0;
    int i_partition_count = 0;
    int i_messageset_size = 0;
    char *p_topic_name = NULL;
    Frlist_node *p_st_node = NULL;
    Kafka_topic_info *p_st_topic = NULL;
    Kafka_partition_info *p_st_partition = NULL;
    Frlist_header st_produce;
    Frlist_header st_messages;
    memset(&st_produce, 0x0, sizeof(Frlist_header));
    L7_req_kafka *p_st_req = p_st_private->p_st_req;
    Frbyte_Packet *p_st_data = p_st_private->p_st_fragment->p_st_data;

    /* get RequiredAcks(if it is 0, server will not send any response) */
    i_ret = bytebuff_get_short(p_st_data, 2, &u16_required_acks, BYTE_BIG_ENDIAN);
    if (i_ret != FR_SUCC)
    {
        frdbg("get RequiredAcks failed!\n");
        return APM_L7_ANALYZE_FAIL;
    }
    /* skip Timeout */
    i_ret = bytebuff_skip(p_st_data, 4);
    if (i_ret != FR_SUCC)
    {
        frdbg("skip Timeout failed!\n");
        return APM_L7_ANALYZE_FAIL;
    }

    /* get array count */
    i_ret = kafka_get_int(p_st_data, &i_topic_count);
    if (i_ret != APM_L7_ANALYZE_SUCC)
    {
        return i_ret;
    }
    else if (i_topic_count < 0)
    {
        return APM_L7_ANALYZE_FAIL;
    }

    /* 解析topics */
    while (i_topic_count--)
    {
        /* get topic name */
        i_ret = kafka_get_string(p_st_data, &p_topic_name);
        if (i_ret != APM_L7_ANALYZE_SUCC)
        {
            frlist_remove_all(&st_produce, kafka_topic_node_free);
            return i_ret;
        }
        /* 创建topic */
        p_st_topic = (Kafka_topic_info *)frmalloc(sizeof(Kafka_topic_info));
        if (p_st_topic == NULL)
        {
            kafka_free(&p_topic_name);
            frlist_remove_all(&st_produce, kafka_topic_node_free);
            return APM_L7_ANALYZE_FAIL;
        }
        memset(p_st_topic, 0x0, sizeof(Kafka_topic_info));
        p_st_topic->p_topic_name = p_topic_name;
        p_topic_name = NULL;

        /* get array count */
        i_ret = kafka_get_int(p_st_data, &i_partition_count);
        if (i_ret != APM_L7_ANALYZE_SUCC)
        {
            kafka_topic_free(&p_st_topic);
            frlist_remove_all(&st_produce, kafka_topic_node_free);
            return i_ret;
        }
        else if (i_partition_count < 0)
        {
            kafka_topic_free(&p_st_topic);
            frlist_remove_all(&st_produce, kafka_topic_node_free);
            return APM_L7_ANALYZE_FAIL;
        }

        /* 解析partitions */
        while (i_partition_count--)
        {
            /* skip partition */
            i_ret = bytebuff_skip(p_st_data, 4);
            if (i_ret != FR_SUCC)
            {
                kafka_topic_free(&p_st_topic);
                frlist_remove_all(&st_produce, kafka_topic_node_free);
                return APM_L7_ANALYZE_FAIL;
            }
            /* get messageset size */
            i_ret = kafka_get_int(p_st_data, &i_messageset_size);
            if (i_ret != APM_L7_ANALYZE_SUCC)
            {
                kafka_topic_free(&p_st_topic);
                frlist_remove_all(&st_produce, kafka_topic_node_free);
                return i_ret;
            }
            /* parse messageset */
            memset(&st_messages, 0x0, sizeof(Frlist_header));
            i_ret = kafka_parse_message_set(p_st_data, i_messageset_size
                                           , &st_messages);
            if (i_ret != APM_L7_ANALYZE_SUCC)
            {
                kafka_topic_free(&p_st_topic);
                frlist_remove_all(&st_produce, kafka_topic_node_free);
                return i_ret;
            }
            /* 创建partition */
            p_st_partition = (Kafka_partition_info *)frmalloc(sizeof(Kafka_partition_info));
            if (p_st_partition == NULL)
            {
                frlist_remove_all(&st_messages, kafka_message_set_node_free);
                kafka_topic_free(&p_st_topic);
                frlist_remove_all(&st_produce, kafka_topic_node_free);
                return APM_L7_ANALYZE_FAIL;
            }
            memset(p_st_partition, 0x0, sizeof(Kafka_partition_info));
            p_st_partition->st_messages = st_messages;

            /* 创建节点 */
            p_st_node = frlist_node_create(p_st_partition);
            if (p_st_node == NULL)
            {
                kafka_partition_free(&p_st_partition);
                kafka_topic_free(&p_st_topic);
                frlist_remove_all(&st_produce, kafka_topic_node_free);
                return APM_L7_ANALYZE_FAIL;
            }
            /* 追加节点 */
            frlist_push(&(p_st_topic->st_partitions), p_st_node);
        }

        /* 创建节点 */
        p_st_node = frlist_node_create(p_st_topic);
        if (p_st_node == NULL)
        {
            kafka_topic_free(&p_st_topic);
            frlist_remove_all(&st_produce, kafka_topic_node_free);
            return APM_L7_ANALYZE_FAIL;
        }
        /* 追加节点 */
        frlist_push(&st_produce, p_st_node);
    }

    /* 格式check */
    if (bytebuff_remaining(p_st_data))
    {
        frlist_remove_all(&st_produce, kafka_topic_node_free);
        return APM_L7_ANALYZE_FAIL;
    }

    if (u16_required_acks == 0)
    {
        p_st_req->i_need_response = FR_NO;
    }
    p_st_req->st_topics = st_produce;
    return APM_L7_ANALYZE_SUCC;
}

/*******************************************************************************
*  FUNC     :  解析fetch request
*  ARGS     :  p_st_private (KAFKA私有数据)
*           :  i_whole_flg(packe是否完整标识)
*  RTN      :  参照module.h
*  NOTE     :
*******************************************************************************/
int __kafka_fetch_req(L7session_kafka_private *p_st_private)
{
    int i_ret = APM_L7_ANALYZE_SUCC;
    int i_topic_count = 0;
    int i_partition_count = 0;
    char *p_topic_name = NULL;
    Frlist_node *p_st_node = NULL;
    Kafka_topic_info *p_st_topic = NULL;
    Frlist_header st_fetch;
    memset(&st_fetch, 0x0, sizeof(Frlist_header));
    L7_req_kafka *p_st_req = p_st_private->p_st_req;
    Frbyte_Packet *p_st_data = p_st_private->p_st_fragment->p_st_data;

    /* skip ReplicaId MaxWaitTime MinBytes */
    i_ret = bytebuff_skip(p_st_data, 4 + 4 + 4);
    if (i_ret != FR_SUCC)
    {
        frdbg("skip ReplicaId and MaxWaitTime and MinBytes failed!\n");
        return APM_L7_ANALYZE_FAIL;
    }

    /* get array count */
    i_ret = kafka_get_int(p_st_data, &i_topic_count);
    if (i_ret != APM_L7_ANALYZE_SUCC)
    {
        return i_ret;
    }
    else if (i_topic_count < 0)
    {
        return APM_L7_ANALYZE_FAIL;
    }

    /* 解析topic */
    while (i_topic_count--)
    {
        /* get topic name */
        i_ret = kafka_get_string(p_st_data, &p_topic_name);
        if (i_ret != APM_L7_ANALYZE_SUCC)
        {
            frdbg("get topic name failed!\n");
            frlist_remove_all(&st_fetch, kafka_topic_node_free);
            return i_ret;
        }
        /* 创建topic */
        p_st_topic = (Kafka_topic_info *)frmalloc(sizeof(Kafka_topic_info));
        if (p_st_topic == NULL)
        {
            kafka_free(&p_topic_name);
            frlist_remove_all(&st_fetch, kafka_topic_node_free);
            return APM_L7_ANALYZE_FAIL;
        }
        memset(p_st_topic, 0x0, sizeof(Kafka_topic_info));
        p_st_topic->p_topic_name = p_topic_name;
        p_topic_name = NULL;

        /* get array count */
        i_ret = kafka_get_int(p_st_data, &i_partition_count);
        if (i_ret != APM_L7_ANALYZE_SUCC)
        {
            frdbg("get array count failed!\n");
            kafka_topic_free(&p_st_topic);
            frlist_remove_all(&st_fetch, kafka_topic_node_free);
            return i_ret;
        }
        else if (i_partition_count < 0)
        {
            kafka_topic_free(&p_st_topic);
            frlist_remove_all(&st_fetch, kafka_topic_node_free);
            return APM_L7_ANALYZE_FAIL;
        }

        /* 解析partition */
        while (i_partition_count--)
        {
            /* skip partition and fetch offset and max bytes */
            i_ret = bytebuff_skip(p_st_data, 4 + 8 + 4);
            if (i_ret != FR_SUCC)
            {
                frdbg("skip partition and fetch offset and max bytes failed!\n");
                return APM_L7_ANALYZE_FAIL;
            }
        }

        /* 创建节点 */
        p_st_node = frlist_node_create(p_st_topic);
        if (p_st_node == NULL)
        {
            kafka_topic_free(&p_st_topic);
            frlist_remove_all(&st_fetch, kafka_topic_node_free);
            return APM_L7_ANALYZE_FAIL;
        }
        /* 追加节点 */
        frlist_push(&st_fetch, p_st_node);
    }

    /* 格式check */
    if (bytebuff_remaining(p_st_data))
    {
        frlist_remove_all(&st_fetch, kafka_topic_node_free);
        return APM_L7_ANALYZE_FAIL;
    }

    p_st_req->st_topics = st_fetch;
    return APM_L7_ANALYZE_SUCC;
}

/*******************************************************************************
*  FUNC     :  解析request
*  ARGS     :  p_st_conn (入力链接信息)
*           :  p_st_hash_report(Frreport_data)
*           :  p_st_private(KAFKA私有数据)
*  RTN      :  参照module.h
*  NOTE     :
*******************************************************************************/
int kafka_parse_request(Conntrack *p_st_conn
                        , Frhash_header *p_st_hash_report
                        , L7session_kafka_private *p_st_private)
{
    int i_ret = APM_L7_ANALYZE_SUCC;
    uint16_t u16_req_type = 0;
    L7_fragment_kafka *p_st_fragment = p_st_private->p_st_fragment;
    Frbyte_Packet *p_st_data = p_st_fragment->p_st_data;

    if (p_st_private->p_st_req)
    {
        frdbg("thres is request before!\n");
        kafka_destroy_request(&p_st_private->p_st_req);
    }
    p_st_private->p_st_req = (L7_req_kafka *)frmalloc(sizeof(L7_req_kafka));
    if (p_st_private->p_st_req == NULL)
    {
        frdbg("create request failed!\n");
        return APM_L7_ANALYZE_FAIL;
    }
    memset(p_st_private->p_st_req, 0x0, sizeof(L7_req_kafka));
    p_st_private->p_st_req->i_need_response = FR_YES;

    /* get request type */
    i_ret = bytebuff_get_short(p_st_data, 2, &u16_req_type, BYTE_BIG_ENDIAN);
    if (i_ret != FR_SUCC)
    {
        frdbg("get request type failed!\n");
        return APM_L7_ANALYZE_FAIL;
    }
    p_st_private->p_st_req->i_req_type = u16_req_type;
    /* skip ApiVersion and CorrelationId */
    i_ret = bytebuff_skip(p_st_data, 2 + 4);
    if (i_ret != FR_SUCC)
    {
        frdbg("skip ApiVersion and CorrelationId failed!\n");
        return APM_L7_ANALYZE_FAIL;
    }
    /* skip ClientId */
    i_ret = kafka_skip_string(p_st_data);
    if (i_ret != APM_L7_ANALYZE_SUCC)
    {
        frdbg("skip ClientId failed!\n");
        return APM_L7_ANALYZE_FAIL;
    }

    /* 根据request type解析 */
    switch (u16_req_type)
    {
        case KAFKA_REQ_TP_PRODUCE:
            i_ret = __kafka_produce_req(p_st_private);
            if (i_ret == APM_L7_ANALYZE_FAIL)
            {
                frdbg("parse produce failed!\n");
                return APM_L7_ANALYZE_FAIL;
            }
            break;

        case KAFKA_REQ_TP_FETCH:
            i_ret = __kafka_fetch_req(p_st_private);
            if (i_ret == APM_L7_ANALYZE_FAIL)
            {
                frdbg("parse fetch failed!\n");
                return APM_L7_ANALYZE_FAIL;
            }
            break;

        /* 业务层不关心的数据，不解析不上报 */
        case KAFKA_REQ_TP_OFFSET:
        case KAFKA_REQ_TP_METADATA:
        case KAFKA_REQ_TP_OFFSET_COMMIT:
        case KAFKA_REQ_TP_OFFSET_FETCH:
        case KAFKA_REQ_TP_CONSUMER_METADATA:
            return APM_L7_ANALYZE_SUCC;

        default:
            frdbg("unknow request type!\n");
            return APM_L7_ANALYZE_FAIL;
    }

    i_ret = __kafka_report_req(p_st_conn, p_st_hash_report, p_st_private);
    /* 不需要resposne响应 */
    if (p_st_private->p_st_req->i_need_response == FR_NO)
    {
        kafka_destroy_request(&p_st_private->p_st_req);
    }
    kafka_init_fragment(p_st_private->p_st_fragment);
    return i_ret;
}


/*******************************************************************************
*  FUNC     :  解析produce request
*  ARGS     :  p_st_private (KAFKA私有数据)
*  RTN      :  APM_L7_ANALYZE_SUCC
*              APM_L7_ANALYZE_FAIL
*  NOTE     :
*******************************************************************************/
int __kafka_produce_req_7(L7session_kafka_private *p_st_private)
{
    int i_ret = APM_L7_ANALYZE_SUCC;
    int i_message_len = 0;
    char *p_topic_name = NULL;
    Frlist_node *p_st_node = NULL;
    Kafka_topic_info *p_st_topic = NULL;
    Kafka_partition_info *p_st_partition = NULL;
    Frlist_header st_produce;
    Frlist_header st_messages;
    memset(&st_produce, 0x0, sizeof(Frlist_header));
    L7_req_kafka *p_st_req = p_st_private->p_st_req;
    Frbyte_Packet *p_st_data = p_st_private->p_st_fragment->p_st_data;

    /* get topic name */
    i_ret = kafka_get_string(p_st_data, &p_topic_name);
    if (i_ret != APM_L7_ANALYZE_SUCC)
    {
        return i_ret;
    }
    /* 创建topic */
    p_st_topic = (Kafka_topic_info *)frmalloc(sizeof(Kafka_topic_info));
    if (p_st_topic == NULL)
    {
        kafka_free(&p_topic_name);
        return APM_L7_ANALYZE_FAIL;
    }
    memset(p_st_topic, 0x0, sizeof(Kafka_topic_info));
    p_st_topic->p_topic_name = p_topic_name;
    p_topic_name = NULL;

    /* skip partition */
    i_ret = bytebuff_skip(p_st_data, 4);
    if (i_ret != FR_SUCC)
    {
        kafka_topic_free(&p_st_topic);
        return APM_L7_ANALYZE_FAIL;
    }
    /* get messages size */
    i_ret = kafka_get_int(p_st_data, &i_message_len);
    if (i_ret != APM_L7_ANALYZE_SUCC)
    {
        kafka_topic_free(&p_st_topic);
        return i_ret;
    }
    if (i_message_len && p_st_private->i_msg_version == 0)
    {
        i_ret = kafka_get_msg_version(p_st_data, i_message_len
                                     , &p_st_private->i_msg_version);
        if (i_ret != APM_L7_ANALYZE_SUCC)
        {
            kafka_topic_free(&p_st_topic);
            return i_ret;
        }
    }

    /* parse messageset */
    memset(&st_messages, 0x0, sizeof(Frlist_header));
    i_ret = kafka_parse_messages(p_st_data, i_message_len
                                , p_st_private->i_msg_version, &st_messages);
    if (i_ret != APM_L7_ANALYZE_SUCC)
    {
        kafka_topic_free(&p_st_topic);
        return i_ret;
    }
    /* 创建partition */
    p_st_partition = (Kafka_partition_info *)frmalloc(sizeof(Kafka_partition_info));
    if (p_st_partition == NULL)
    {
        frlist_remove_all(&st_messages, kafka_message_set_node_free);
        kafka_topic_free(&p_st_topic);
        return APM_L7_ANALYZE_FAIL;
    }
    memset(p_st_partition, 0x0, sizeof(Kafka_partition_info));
    p_st_partition->st_messages = st_messages;

    /* 创建节点 */
    p_st_node = frlist_node_create(p_st_partition);
    if (p_st_node == NULL)
    {
        kafka_partition_free(&p_st_partition);
        kafka_topic_free(&p_st_topic);
        return APM_L7_ANALYZE_FAIL;
    }
    /* 追加节点 */
    frlist_push(&(p_st_topic->st_partitions), p_st_node);

    /* 创建节点 */
    p_st_node = frlist_node_create(p_st_topic);
    if (p_st_node == NULL)
    {
        kafka_topic_free(&p_st_topic);
        return APM_L7_ANALYZE_FAIL;
    }
    /* 追加节点 */
    frlist_push(&st_produce, p_st_node);

    /* 格式check */
    if (bytebuff_remaining(p_st_data))
    {
        frlist_remove_all(&st_produce, kafka_topic_node_free);
        return APM_L7_ANALYZE_FAIL;
    }

    p_st_req->i_need_response = FR_NO;
    p_st_req->st_topics = st_produce;
    return APM_L7_ANALYZE_SUCC;
}

/*******************************************************************************
*  FUNC     :  解析produce request
*  ARGS     :  p_st_private (KAFKA私有数据)
*  RTN      :  APM_L7_ANALYZE_SUCC
*              APM_L7_ANALYZE_FAIL
*  NOTE     :
*******************************************************************************/
int __kafka_multiproduce_req_7(L7session_kafka_private *p_st_private)
{
    int i_ret = APM_L7_ANALYZE_SUCC;
    int i_message_len = 0;
    uint16_t u16_topic_count = 0;
    char *p_topic_name = NULL;
    Frlist_node *p_st_node = NULL;
    Kafka_topic_info *p_st_topic = NULL;
    Kafka_partition_info *p_st_partition = NULL;
    Frlist_header st_produce;
    Frlist_header st_messages;
    memset(&st_produce, 0x0, sizeof(Frlist_header));
    L7_req_kafka *p_st_req = p_st_private->p_st_req;
    Frbyte_Packet *p_st_data = p_st_private->p_st_fragment->p_st_data;

    /* 取得topic count */
    i_ret = bytebuff_get_short(p_st_data, 2, &u16_topic_count, BYTE_BIG_ENDIAN);
    if (i_ret != FR_SUCC)
    {
        frdbg("get topic count failed!\n");
        return APM_L7_ANALYZE_FAIL;
    }

    while (u16_topic_count--)
    {
        /* get topic name */
        i_ret = kafka_get_string(p_st_data, &p_topic_name);
        if (i_ret != APM_L7_ANALYZE_SUCC)
        {
            frlist_remove_all(&st_produce, kafka_topic_node_free);
            return i_ret;
        }
        /* 创建topic */
        p_st_topic = (Kafka_topic_info *)frmalloc(sizeof(Kafka_topic_info));
        if (p_st_topic == NULL)
        {
            kafka_free(&p_topic_name);
            frlist_remove_all(&st_produce, kafka_topic_node_free);
            return APM_L7_ANALYZE_FAIL;
        }
        memset(p_st_topic, 0x0, sizeof(Kafka_topic_info));
        p_st_topic->p_topic_name = p_topic_name;
        p_topic_name = NULL;

        /* skip partition */
        i_ret = bytebuff_skip(p_st_data, 4);
        if (i_ret != FR_SUCC)
        {
            kafka_topic_free(&p_st_topic);
            frlist_remove_all(&st_produce, kafka_topic_node_free);
            return APM_L7_ANALYZE_FAIL;
        }
        /* get messages size */
        i_ret = kafka_get_int(p_st_data, &i_message_len);
        if (i_ret != APM_L7_ANALYZE_SUCC)
        {
            kafka_topic_free(&p_st_topic);
            frlist_remove_all(&st_produce, kafka_topic_node_free);
            return i_ret;
        }
        if (i_message_len && p_st_private->i_msg_version == 0)
        {
            i_ret = kafka_get_msg_version(p_st_data, i_message_len
                                         , &p_st_private->i_msg_version);
            if (i_ret != APM_L7_ANALYZE_SUCC)
            {
                kafka_topic_free(&p_st_topic);
                frlist_remove_all(&st_produce, kafka_topic_node_free);
                return i_ret;
            }
        }

        /* parse messageset */
        memset(&st_messages, 0x0, sizeof(Frlist_header));
        i_ret = kafka_parse_messages(p_st_data, i_message_len
                                    , p_st_private->i_msg_version, &st_messages);
        if (i_ret != APM_L7_ANALYZE_SUCC)
        {
            kafka_topic_free(&p_st_topic);
            frlist_remove_all(&st_produce, kafka_topic_node_free);
            return i_ret;
        }
        /* 创建partition */
        p_st_partition = (Kafka_partition_info *)frmalloc(sizeof(Kafka_partition_info));
        if (p_st_partition == NULL)
        {
            frlist_remove_all(&st_messages, kafka_message_set_node_free);
            kafka_topic_free(&p_st_topic);
            frlist_remove_all(&st_produce, kafka_topic_node_free);
            return APM_L7_ANALYZE_FAIL;
        }
        memset(p_st_partition, 0x0, sizeof(Kafka_partition_info));
        p_st_partition->st_messages = st_messages;

        /* 创建节点 */
        p_st_node = frlist_node_create(p_st_partition);
        if (p_st_node == NULL)
        {
            kafka_partition_free(&p_st_partition);
            kafka_topic_free(&p_st_topic);
            frlist_remove_all(&st_produce, kafka_topic_node_free);
            return APM_L7_ANALYZE_FAIL;
        }
        /* 追加节点 */
        frlist_push(&(p_st_topic->st_partitions), p_st_node);

        /* 创建节点 */
        p_st_node = frlist_node_create(p_st_topic);
        if (p_st_node == NULL)
        {
            kafka_topic_free(&p_st_topic);
            frlist_remove_all(&st_produce, kafka_topic_node_free);
            return APM_L7_ANALYZE_FAIL;
        }
        /* 追加节点 */
        frlist_push(&st_produce, p_st_node);
    }

    /* 格式check */
    if (bytebuff_remaining(p_st_data))
    {
        frlist_remove_all(&st_produce, kafka_topic_node_free);
        return APM_L7_ANALYZE_FAIL;
    }

    p_st_req->i_need_response = FR_NO;
    p_st_req->st_topics = st_produce;
    return APM_L7_ANALYZE_SUCC;
}

/*******************************************************************************
*  FUNC     :  解析fetch request
*  ARGS     :  p_st_private (KAFKA私有数据)
*  RTN      :  APM_L7_ANALYZE_SUCC
*              APM_L7_ANALYZE_FAIL
*  NOTE     :
*******************************************************************************/
int __kafka_fetch_req_7(L7session_kafka_private *p_st_private)
{
    int i_ret = APM_L7_ANALYZE_SUCC;
    char *p_topic_name = NULL;
    Frlist_node *p_st_node = NULL;
    Kafka_topic_info *p_st_topic = NULL;
    Frlist_header st_produce;
    memset(&st_produce, 0x0, sizeof(Frlist_header));
    L7_req_kafka *p_st_req = p_st_private->p_st_req;
    Frbyte_Packet *p_st_data = p_st_private->p_st_fragment->p_st_data;

    /* get topic name */
    i_ret = kafka_get_string(p_st_data, &p_topic_name);
    if (i_ret != APM_L7_ANALYZE_SUCC)
    {
        return i_ret;
    }
    /* 创建topic */
    p_st_topic = (Kafka_topic_info *)frmalloc(sizeof(Kafka_topic_info));
    if (p_st_topic == NULL)
    {
        kafka_free(&p_topic_name);
        return APM_L7_ANALYZE_FAIL;
    }
    memset(p_st_topic, 0x0, sizeof(Kafka_topic_info));
    p_st_topic->p_topic_name = p_topic_name;
    p_topic_name = NULL;

    /* skip partition OFFSET OFFSET */
    i_ret = bytebuff_skip(p_st_data, 4 + 8 + 4);
    if (i_ret != FR_SUCC)
    {
        kafka_topic_free(&p_st_topic);
        return APM_L7_ANALYZE_FAIL;
    }

    /* 创建节点 */
    p_st_node = frlist_node_create(p_st_topic);
    if (p_st_node == NULL)
    {
        kafka_topic_free(&p_st_topic);
        return APM_L7_ANALYZE_FAIL;
    }
    /* 追加节点 */
    frlist_push(&st_produce, p_st_node);

    /* 格式check */
    if (bytebuff_remaining(p_st_data))
    {
        frlist_remove_all(&st_produce, kafka_topic_node_free);
        return APM_L7_ANALYZE_FAIL;
    }

    p_st_req->st_topics = st_produce;
    return APM_L7_ANALYZE_SUCC;
}

/*******************************************************************************
*  FUNC     :  解析fetch request
*  ARGS     :  p_st_private (KAFKA私有数据)
*  RTN      :  APM_L7_ANALYZE_SUCC
*              APM_L7_ANALYZE_FAIL
*  NOTE     :
*******************************************************************************/
int __kafka_multifetch_req_7(L7session_kafka_private *p_st_private)
{
    int i_ret = APM_L7_ANALYZE_SUCC;
    uint16_t u16_topic_count = 0;
    char *p_topic_name = NULL;
    Frlist_node *p_st_node = NULL;
    Kafka_topic_info *p_st_topic = NULL;
    Frlist_header st_produce;
    memset(&st_produce, 0x0, sizeof(Frlist_header));
    L7_req_kafka *p_st_req = p_st_private->p_st_req;
    Frbyte_Packet *p_st_data = p_st_private->p_st_fragment->p_st_data;

    /* 取得topic count */
    i_ret = bytebuff_get_short(p_st_data, 2, &u16_topic_count, BYTE_BIG_ENDIAN);
    if (i_ret != FR_SUCC)
    {
        frdbg("get topic count failed!\n");
        return APM_L7_ANALYZE_FAIL;
    }

    while (u16_topic_count--)
    {
        /* get topic name */
        i_ret = kafka_get_string(p_st_data, &p_topic_name);
        if (i_ret != APM_L7_ANALYZE_SUCC)
        {
            frlist_remove_all(&st_produce, kafka_topic_node_free);
            return i_ret;
        }
        /* 创建topic */
        p_st_topic = (Kafka_topic_info *)frmalloc(sizeof(Kafka_topic_info));
        if (p_st_topic == NULL)
        {
            kafka_free(&p_topic_name);
            frlist_remove_all(&st_produce, kafka_topic_node_free);
            return APM_L7_ANALYZE_FAIL;
        }
        memset(p_st_topic, 0x0, sizeof(Kafka_topic_info));
        p_st_topic->p_topic_name = p_topic_name;
        p_topic_name = NULL;

        /* skip partition OFFSET OFFSET */
        i_ret = bytebuff_skip(p_st_data, 4 + 8 + 4);
        if (i_ret != FR_SUCC)
        {
            kafka_topic_free(&p_st_topic);
            frlist_remove_all(&st_produce, kafka_topic_node_free);
            return APM_L7_ANALYZE_FAIL;
        }

        /* 创建节点 */
        p_st_node = frlist_node_create(p_st_topic);
        if (p_st_node == NULL)
        {
            kafka_topic_free(&p_st_topic);
            frlist_remove_all(&st_produce, kafka_topic_node_free);
            return APM_L7_ANALYZE_FAIL;
        }
        /* 追加节点 */
        frlist_push(&st_produce, p_st_node);
    }

    /* 格式check */
    if (bytebuff_remaining(p_st_data))
    {
        frlist_remove_all(&st_produce, kafka_topic_node_free);
        return APM_L7_ANALYZE_FAIL;
    }

    p_st_req->st_topics = st_produce;
    return APM_L7_ANALYZE_SUCC;
}

/*******************************************************************************
*  FUNC     :  解析request
*  ARGS     :  p_st_conn (入力链接信息)
*           :  p_st_hash_report(Frreport_data)
*           :  p_st_private(KAFKA私有数据)
*  RTN      :  参照module.h
*  NOTE     :
*******************************************************************************/
int kafka_parse_request_7(Conntrack *p_st_conn
                         , Frhash_header *p_st_hash_report
                         , L7session_kafka_private *p_st_private)
{
    int i_ret = APM_L7_ANALYZE_SUCC;
    uint16_t u16_req_type = 0;
    L7_fragment_kafka *p_st_fragment = p_st_private->p_st_fragment;
    Frbyte_Packet *p_st_data = p_st_fragment->p_st_data;

    if (p_st_private->p_st_req)
    {
        frdbg("thres is request before!\n");
        kafka_destroy_request(&p_st_private->p_st_req);
    }
    p_st_private->p_st_req = (L7_req_kafka *)frmalloc(sizeof(L7_req_kafka));
    if (p_st_private->p_st_req == NULL)
    {
        frdbg("create request failed!\n");
        return APM_L7_ANALYZE_FAIL;
    }
    memset(p_st_private->p_st_req, 0x0, sizeof(L7_req_kafka));
    p_st_private->p_st_req->i_need_response = FR_YES;

    /* get request type */
    i_ret = bytebuff_get_short(p_st_data, 2, &u16_req_type, BYTE_BIG_ENDIAN);
    if (i_ret != FR_SUCC)
    {
        frdbg("get request type failed!\n");
        return APM_L7_ANALYZE_FAIL;
    }
    p_st_private->p_st_req->i_req_type = u16_req_type;

    switch (u16_req_type)
    {
        case KAFKA_REQ_TP_PRODUCE:
            i_ret = __kafka_produce_req_7(p_st_private);
            break;

        case KAFKA_REQ_TP_FETCH:
            i_ret = __kafka_fetch_req_7(p_st_private);
            break;

        case KAFKA_REQ_TP_MULTIFETCH:
            //i_ret = __kafka_multifetch_req_7(p_st_private);
            break;

        case KAFKA_REQ_TP_MULTIPRODUCE:
            i_ret = __kafka_multiproduce_req_7(p_st_private);
            break;

        case KAFKA_REQ_TP_OFFSETS:
            return APM_L7_ANALYZE_SUCC;

        default:
            frdbg("unknow request type!\n");
            return APM_L7_ANALYZE_FAIL;
    }

    i_ret = __kafka_report_req(p_st_conn, p_st_hash_report, p_st_private);
    /* 不需要resposne响应 */
    if (p_st_private->p_st_req->i_need_response == FR_NO)
    {
        kafka_destroy_request(&p_st_private->p_st_req);
    }
    kafka_init_fragment(p_st_private->p_st_fragment);
    return i_ret;
}
